package com.qch.rocket.demo.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Set;

/**
 * @Author qch
 * @Date 2021/6/7
 * 拉取消息的消费者
 */
public class MyConsumer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException, UnsupportedEncodingException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consuer_grp_01");
        consumer.setNamesrvAddr("192.168.142.134:9876");
        consumer.start();
        Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues("tp_demo_01");
       //遍历该主题的谷歌消息队列，进行消费
        for (MessageQueue messageQueue : messageQueues) {
            /**
             * 1.表示当前主题的一个消息队列
             * 2.一个表达式，对接收的消息按照tag进行过滤
             * 支持tag1||tag2||*(表示不敌消息进行tag过滤)
             * 3.消息的偏移量，从这里开始消费
             * 4.每次最多拉取多少条
             */
            PullResult result = consumer.pull(messageQueue, "*", 0, 10);
            List<MessageExt> msgFoundList = result.getMsgFoundList();
            if (msgFoundList==null)continue;
            System.out.println("*****************message_quee**********"+messageQueue);
            for (MessageExt messageExt : msgFoundList) {
                System.out.println(messageExt);
                System.out.println(new String(messageExt.getBody(),"utf-8"));
            }

        }

        consumer.shutdown();
    }
}
